package com.snopzyz.hello.databus.domain;

import com.github.shyiko.mysql.binlog.event.*;
import com.snopzyz.hello.databus.enums.ColumnKeyType;
import com.snopzyz.hello.databus.enums.DmlType;
import com.snopzyz.hello.databus.utils.BinlogUtils;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.io.Serializable;
import java.util.*;
import java.util.stream.Collectors;

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
public class BinlogRawData {

    /**
     * 变更类型：INSERT、DELETE、UPDATE
     */
    private DmlType dmlType;

    /**
     * 库名
     */
    private String database;

    /**
     * 表名
     */
    private String table;

    /**
     * binlog所属的从库服务器id
     */
    private Long serverId;

    /**
     * 变更数据差异，key=列名，value=变更前后数据和字段信息
     */
    private Map<String, ColumnChangeInfo> diffMap;

    /**
     * 产生时间
     */
    private Long timestamp;


    /**
     * 实体字段变更信息
     */
    @Data
    @Builder
    static class ColumnChangeInfo {
        /**
         * 字段索引类型：主键、唯一索引、普通索引、非索引
         */
        private ColumnKeyType keyType;
        /**
         * 字段数据类型 如"varchar(256)"、"bigint"...
         */
        private String dataType;
        /**
         * 变更前的值
         */
        private String oldValue;
        /**
         * 变更后的值
         */
        private String newValue;
    }



    public static List<BinlogRawData> from (Event event, String dbTable, Map<String, Column> cols) {
        EventType eventType = event.getHeader().getEventType();
        if (EventType.isWrite(eventType)) {
            WriteRowsEventData writeData = event.getData();
            return fromInsertOrDelete(writeData.getRows(), dbTable, cols, event.getHeader());

        } else if (EventType.isUpdate(eventType)) {
            UpdateRowsEventData updateData = event.getData();
            return fromUpdate(updateData.getRows(), dbTable, cols, event.getHeader());

        } else if (EventType.isDelete(eventType)) {
            DeleteRowsEventData deleteData = event.getData();
            return BinlogRawData.fromInsertOrDelete(deleteData.getRows(), dbTable, cols, event.getHeader());
        }
        return new ArrayList<>();
    }


    public static List<BinlogRawData> fromInsertOrDelete (List<Serializable[]> rows, String dbTable, Map<String, Column> cols, EventHeader header) {
        return rows.stream()
                .map(row -> fromInsertOrDelete(row, dbTable, cols, header))
                .collect(Collectors.toList());
    }


    public static BinlogRawData fromInsertOrDelete (Serializable[] row, String dbTable, Map<String, Column> cols, EventHeader header) {
        boolean isInsert = EventType.isWrite(header.getEventType());
        BinlogRawData rawData = fromHeader(header, dbTable);

        Map<String, ColumnChangeInfo> diffMap = new HashMap<>();
        cols.forEach((key, value) -> {
            ColumnChangeInfo changeInfo = ColumnChangeInfo.builder()
                    .keyType(value.getKeyType())
                    .dataType(value.getDataType())
                    .build();
            if (isInsert) {
                changeInfo.setNewValue(Objects.toString(row[value.getIdx()], null));
            } else {
                changeInfo.setOldValue(Objects.toString(row[value.getIdx()], null));
            }
            diffMap.put(key, changeInfo);
        });
        rawData.setDiffMap(diffMap);
        return rawData;
    }


    private static List<BinlogRawData> fromUpdate (List<Map.Entry<Serializable[], Serializable[]>> rows, String dbTable,
                                                   Map<String, Column> cols, EventHeader header) {
        return rows.stream()
                .map(row -> fromUpdate(row, dbTable, cols, header))
                .collect(Collectors.toList());
    }


    public static BinlogRawData fromUpdate (Map.Entry<Serializable[], Serializable[]> row, String dbTable, Map<String, Column> cols, EventHeader header) {
        BinlogRawData rawData = fromHeader(header, dbTable);

        Map<String, ColumnChangeInfo> diffMap = new HashMap<>();
        cols.forEach((key, value) -> {
            ColumnChangeInfo changeInfo = ColumnChangeInfo.builder()
                    .keyType(value.getKeyType())
                    .dataType(value.getDataType())
                    .oldValue(Objects.toString(row.getKey()[value.getIdx()], null))
                    .newValue(Objects.toString(row.getValue()[value.getIdx()], null))
                    .build();
            diffMap.put(key, changeInfo);
        });
        rawData.setDiffMap(diffMap);
        return rawData;
    }

    public static BinlogRawData fromHeader (EventHeader header, String dbTable) {
        return BinlogRawData.builder()
                .dmlType(DmlType.getByEventType(header.getEventType()))
                .database(BinlogUtils.getDb(dbTable))
                .table(BinlogUtils.getTable(dbTable))
                .serverId(header.getServerId())
                .timestamp(header.getTimestamp())
                .build();
    }
}
